#ifndef __ASIO_TCP_CONNECTION_HPP__
#define __ASIO_TCP_CONNECTION_HPP__

#include "../io_session.h"
#include "../tcp_session.h"
#include "../io_filter_chain.h"
#include "../tcp_acceptor_config.h"
#include "../tcp_connection_manager.h"
#include "../tcp_message_context.h"
#include "../io_handler.h"
#include "../tcp_session_config.h"

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/tuple/tuple.hpp>
#include <boost/scoped_ptr.hpp>

#define TCP_DATA_BUFFER_LENGTH   8192
#define DEFAULT_FREE_BUFFER_LENGTH  64

namespace vpm 
{
	namespace mina 
	{
		using namespace boost::asio;
		template <class Session = TcpSession,class SessionConfigPtr = TcpSessionConfigPtr >
			class AsioTcpConnection
			{
				public:
					AsioTcpConnection(IoHandler & handler,TcpAcceptorConfigPtr  acceptConfig)
						:m_ioHandler(&handler),m_tcpAcceptorConfig(acceptConfig)
					{
						m_dataLen =0; 
						m_dataPos =0;
						Session *pSession = (Session*)m_ioHandler->template create_session<Session,SessionConfigPtr>
							(acceptConfig->get_session_config());

						m_ioSession.reset(pSession);
						m_ioHandler->sesion_created(*m_ioSession);
						m_bClosed = false; 
						IoSessionAccessor::attach_writter(*m_ioSession,boost::bind(&AsioTcpConnection::send_data,this,_1));
						IoSessionAccessor::attach_reader(*m_ioSession,boost::bind(&AsioTcpConnection::recv_data,this,_1));
						IoSessionAccessor::attach_closer(*m_ioSession,boost::bind(&AsioTcpConnection::close_connection,this));
					}

					void open()
					{
						//printf("connection opened , start to read some \n");
                        m_ioSession->set_connect_state(true);

						m_ioSession->get_session_config()->remote_host(m_pTcpSocket->remote_endpoint().address().to_string()); 
						m_ioSession->get_session_config()->remote_port(m_pTcpSocket->remote_endpoint().port()); 
						m_ioHandler->session_opened(*m_ioSession);


						m_pTcpSocket->async_read_some(boost::asio::buffer(m_dataBuf,TCP_DATA_BUFFER_LENGTH),
								boost::bind(&AsioTcpConnection::handle_read,this,
									boost::asio::placeholders::error,
									boost::asio::placeholders::bytes_transferred) );
					}

					virtual ~AsioTcpConnection()
					{
						//printf("Deconstruct AsioTcpConnection %p\n",this); 
					}

					int send_data(MessageContext & msg)
					{
						TcpMessageContext & tcpMsg = (TcpMessageContext&) msg;
						return this->send_data(tcpMsg.buffer,tcpMsg.length);		
					}

					virtual int send_data(const char * pData, int dataLen)
					{		
						//printf("Send data :%s\n", pData);
						boost::asio::async_write(*m_pTcpSocket,boost::asio::buffer(pData,dataLen), 
								boost::bind(&AsioTcpConnection::handle_write, 
									this,boost::asio::placeholders::error) );
						return 0; 
					}

					int recv_data(MessageContext & msg)
					{
						TcpMessageContext & tcpMsg = (TcpMessageContext&) msg;
						return this->recv_data(tcpMsg.buffer,tcpMsg.length);
					}

					virtual int recv_data(char * pBuf, int bufLen)
					{
						m_pTcpSocket->async_read_some(boost::asio::buffer(pBuf,bufLen),
								boost::bind(&AsioTcpConnection::handle_read,this,
									boost::asio::placeholders::error,
									boost::asio::placeholders::bytes_transferred)); 
						return 0; 
					}
					int close_connection()
					{
						this->close(); 
						return 0; 
					}

					void create_socket(boost::asio::io_service & ioService)
					{
						m_pTcpSocket.reset( new ip::tcp::socket(ioService));
					}

					ip::tcp::socket & socket()
					{
						return *m_pTcpSocket; 
					}

					void handle_read(const boost::system::error_code & errcode, 
							std::size_t bytes_transferred)
					{
						if (!errcode)
						{
							//printf("received data,length  %d\n",bytes_transferred);
							IoFilterChain & inChain = m_tcpAcceptorConfig->get_input_chain();
							
							m_dataLen += bytes_transferred; 

							int pkgLen = m_ioHandler->message_demarcation(*m_ioSession,m_dataBuf+m_dataPos,m_dataLen); 
							while  (pkgLen > 0 && m_dataLen >= pkgLen)
							{
								TcpMessageContext  tcpMsg(m_dataBuf+m_dataPos,pkgLen); 

								FilterAction act = inChain.process(tcpMsg);
								if (act != FILTER_SKIP_ALL_CHAINS)
								{
									m_ioHandler->message_received(*m_ioSession,tcpMsg);
								}
								m_dataPos += pkgLen; 
								m_dataLen -= pkgLen; 

								if (m_dataLen  ==0)
								{
									m_dataPos =0; 
									break; 
								}
								else if (m_dataPos >= (TCP_DATA_BUFFER_LENGTH - DEFAULT_FREE_BUFFER_LENGTH))
								{
									memmove(m_dataBuf,m_dataBuf+m_dataPos,m_dataLen); 
									m_dataPos =0; 
									break; 
								}

								pkgLen = m_ioHandler->message_demarcation(*m_ioSession,m_dataBuf+m_dataPos,m_dataLen); 
							}

							if (pkgLen ==-1 || m_dataLen >= TCP_DATA_BUFFER_LENGTH )
							{
								//package error 
								m_dataLen = 0; 
								m_dataPos =0; 									
							}

							//else if (pkgLen ==0)
							//{
								//not complete 
							//}

							m_pTcpSocket->async_read_some(boost::asio::buffer(m_dataBuf+m_dataPos,TCP_DATA_BUFFER_LENGTH - m_dataPos),
									boost::bind(&AsioTcpConnection::handle_read,this,
										boost::asio::placeholders::error,
										boost::asio::placeholders::bytes_transferred) );

						}
						else 
						{
							this->close(); 
						}
					}

					void handle_write(const boost::system::error_code & errcode)
					{
						if (!errcode)
						{

						}
						else if (errcode == boost::asio::error::operation_aborted)
						{
							this->close(); 
						}
					}

					void close()
					{
						if (!m_bClosed)// avoid dead loop for use may call close when on sesssion closed
						{
							m_ioSession->set_connect_state(false); 
							m_ioHandler->session_closed(*m_ioSession); 
							if (m_pTcpSocket)
							{
								m_pTcpSocket->close(); 
							}
							m_bClosed = true; 
						}
					}

				private:
					//boost::array<char , TCP_DATA_BUFFER_LENGTH> m_recvBuf; 

					bool m_bClosed; 

					boost::scoped_ptr<ip::tcp::socket> m_pTcpSocket; 
					char m_dataBuf[TCP_DATA_BUFFER_LENGTH]; 
					int m_dataPos; 
					int m_dataLen; 
					IoHandler * m_ioHandler; 
					TcpAcceptorConfigPtr  m_tcpAcceptorConfig; 	
					std::auto_ptr<IoSession> m_ioSession;
					SessionConfigPtr m_sessionConfig;
			};
#define TcpConnectionImpl AsioTcpConnection
	}
}

#endif // 

